package com.gitee.loyo;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.streams.Pump;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class Jumpers {
    private final static Logger log = LoggerFactory.getLogger(Jumpers.class);
    public final static Map<Integer, NetServer> jumperServers = new ConcurrentHashMap<>();

    public static Future<Void> startup(Vertx vertx, Integer port){
        shutdown(port);
        if(!contain(port)){
            return Future.failedFuture("not found jumper");
        }

        Jumpers.Jumper jumper = get(port);
        NetClient netClient = vertx.createNetClient();
        return vertx.createNetServer(new NetServerOptions().setPort(port)).exceptionHandler(t ->
            log.error("net server 127.0.0.1:" + port + " error", t)
        ).connectHandler(serverSocket -> {
            log.info("new connection connect to 127.0.0.1:{}", port);
            serverSocket.pause().exceptionHandler(t ->
                log.error("127.0.0.1:" + port + " connection  error", t)
            ).closeHandler(avoid ->
                log.info("127.0.0.1:" + port + " connection close")
            );
            netClient.connect(jumper.remotePort, jumper.remoteHost).onComplete(ar -> {
                if(ar.failed()){
                    log.error("failed to connect to {}:{}", jumper.remoteHost, jumper.remotePort);
                    serverSocket.close();
                    return;
                }

                NetSocket clientSocket = ar.result();
                clientSocket.pause().exceptionHandler(t ->
                    log.error(jumper.remoteHost + ":" + jumper.remotePort + " connection error", t)
                ).closeHandler(avoid ->
                    log.info(jumper.remoteHost + ":" + jumper.remotePort + " connection close")
                );
                Pump.pump(serverSocket, clientSocket).start();
                Pump.pump(clientSocket, serverSocket).start();
                clientSocket.resume();
                serverSocket.resume();
            });
        }).listen().onSuccess(netServer ->
            jumperServers.put(port, netServer)
        ).map((Void)null);
    }

    public static Future<Void> shutdown(Integer port){
        if(jumperServers.containsKey(port)){
            return jumperServers.remove(port).close();
        }
        return Future.succeededFuture();
    }

    public static boolean contain(int port){
        return BridgeConfig.INSTANCE.jumpers.stream().filter(jumper -> jumper.port == port).count() > 0;
    }

    public static Jumper get(Integer port){
        return BridgeConfig.INSTANCE.jumpers.stream().filter(jumper -> jumper.port == port).findFirst().get();
    }

    public static Future<Void> add(Jumper jumper){
        synchronized (BridgeConfig.INSTANCE.jumpers){
            if(contain(jumper.port)){
                return Future.failedFuture("port is exists");
            }

            BridgeConfig.INSTANCE.jumpers.add(jumper);
        }

        return BridgeConfig.INSTANCE.save();
    }

    public static Future<Void> update(Jumper jumper){
        synchronized (BridgeConfig.INSTANCE.jumpers){
            if(!contain(jumper.port)){
                return Future.failedFuture("port is not exists");
            }
            Jumpers.Jumper record = get(jumper.port);
            record.remoteHost = jumper.remoteHost;
            record.remotePort = jumper.remotePort;
            record.note = jumper.note;
        }

        return BridgeConfig.INSTANCE.save().onSuccess(event -> {
            if(jumperServers.containsKey(jumper.port)){
                startup(Vertx.currentContext().owner(), jumper.port);
            }
        });
    }

    public static Future<Void> remove(Integer port){
        synchronized (BridgeConfig.INSTANCE.jumpers){
            if(!contain(port)){
                return Future.failedFuture("port is not exists");
            }
            Iterator<Jumpers.Jumper> ite = BridgeConfig.INSTANCE.jumpers.iterator();
            while (ite.hasNext()){
                Jumpers.Jumper record = ite.next();
                if(record.port == port){
                    ite.remove();
                    break;
                }
            }
        }

        return BridgeConfig.INSTANCE.save().onSuccess(event -> {
            shutdown(port);
        });
    }

    public static class Jumper {
        public final int port;
        public String remoteHost;
        public int remotePort;
        public String note;

        @JsonCreator
        public Jumper(@JsonProperty("port") int port, @JsonProperty("remoteHost") String remoteHost,
                      @JsonProperty("remotePort") int remotePort, @JsonProperty("note") String note) {
            this.port = port;
            this.remoteHost = remoteHost;
            this.remotePort = remotePort;
            this.note = note == null || note.isEmpty() ? null : note.substring(0, Math.min(64, note.length()));
        }

        public boolean running(){
            return jumperServers.containsKey(port);
        }

        public int getPort() {
            return port;
        }

        public String getRemoteHost() {
            return remoteHost;
        }

        public int getRemotePort() {
            return remotePort;
        }

        public String getNote() {
            return note;
        }
    }
}
